package com.wish233.wiliwilivideo.job;

import com.wish233.wiliwilivideo.domain.po.Video;
import com.wish233.wiliwilivideo.mq.MqProducer;
import com.wish233.wiliwilivideo.service.VideoService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @author WHH
 * 同步redis数据
 */
@Service
@RequiredArgsConstructor
@Slf4j
public class ViewSyncScheduler {
    private final StringRedisTemplate redisTemplate;
    private final VideoService videoService;
    private final MqProducer mqProducer;
    // 每 10 秒同步一次
    @Scheduled(fixedDelay = 10_000)
    public void syncViewsToDatabase() {
        // 使用 SCAN 获取所有 video:views:* 的键
        String pattern = "video:views:*";
        ScanOptions options = ScanOptions.scanOptions().match(pattern).count(100).build();
        // 批量收集待处理数据
        Map<String, Long> videoViewsMap = new HashMap<>();
        List<String> keysToClear = new ArrayList<>();
        // 第一次遍历：收集所有需要更新的数据
        try (Cursor<String> cursor = redisTemplate.scan(options)) {
            while (cursor.hasNext()) {
                String key = cursor.next();
                if (key.contains("limiter")) {
                    continue;
                }
                try {
                    String videoId = key.split(":")[2];
                    Long views = Long.valueOf(Objects.requireNonNull(redisTemplate.opsForValue().get(key)));
                    videoViewsMap.put(videoId, views);
                    keysToClear.add(key);
                } catch (Exception e) {
                    log.error("收集键 {} 时出错: {}", key, e.getMessage());
                }
            }
        }
        if (videoViewsMap.isEmpty()) {
            return;
        }
        // 批量查询数据库
        List<Video> videos = videoService.listByIds(videoViewsMap.keySet());
        Map<Long, Video> videoMap = videos.stream()
                .collect(Collectors.toMap(Video::getId, Function.identity()));
        // 准备批量更新数据和发送MQ
        List<Video> videosToUpdate = new ArrayList<>();
        videoViewsMap.forEach((videoId, views) -> {
            Video video = videoMap.get(Long.parseLong(videoId));
            if (video != null) {
                if (views == 0) {
                    return;
                }
                video.setViewCount(video.getViewCount() + views);
                videosToUpdate.add(video);
                mqProducer.send2ES(video, 1);
            }
        });
        if (!videosToUpdate.isEmpty()) {
            // 批量更新数据库
            videoService.updateBatchById(videosToUpdate);
            // 批量清除Redis计数
            redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
                for (String key : keysToClear) {
                    Long delta = videoViewsMap.get(key.split(":")[2]);
                    connection.decrBy(key.getBytes(), delta);
                }
                return null;
            });
        }
    }


}